package com.sc.nft.dao;

import cn.hutool.core.date.DateField;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.sc.nft.entity.*;
import com.sc.nft.entity.dto.MaintainUserConSignCollectionPageDTO;
import com.sc.nft.entity.dto.MarketConsignPageDTO;
import com.sc.nft.entity.dto.SecondConSignCollectionDTO;
import com.sc.nft.entity.vo.ConsignmentSalesIdGetParticularsVO;
import com.sc.nft.entity.vo.MarketConsignPageV3_7VO;
import com.sc.nft.entity.vo.MyConsignmentListVO;
import com.sc.nft.entity.vo.MySalestListVO;
import com.sc.nft.enums.ConSignStatus;
import com.sc.nft.enums.MarketTypeEnum;
import com.sc.nft.enums.RedisKeyEnum;
import com.sc.nft.mapper.SecondCollectionMapper;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.lang.NonNull;
import org.springframework.stereotype.Repository;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.event.TransactionPhase;
import org.springframework.transaction.event.TransactionalEventListener;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.io.IOException;
import java.math.BigDecimal;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * 寄售表(sc_second_collection)数据DAO
 * <p>
 * 寄售表重构：MySQL sc_second_collection 表保留近2个月数据（ STATUS IN (1, 4) 的数据全量保留 ） + 全量数据同步迁移到ES集群中
 *
 * @author kancy
 * @description 由 Mybatisplus Code Generator 创建
 * @since 2022-10-05 00:23:53
 */
@Slf4j
@Repository
@RequiredArgsConstructor
public class SecondCollectionDao {

    private final SecondCollectionEsDao esDao;
    private final SecondCollectionDaoAdapter baseDao;
    private final SecondCollectionMapper baseMapper;
    private final DigitalCollectionDao digitalCollectionDao;
    private final UserCollectionDao userCollectionDao;
    private final ExitConsensusUserDao exitConsensusUserDao;
    private final ApplicationEventPublisher publisher;
    private final UserInfoDao userInfoDao;

    @Resource
    private RedisTemplate redisTemplate;

    public SecondCollection getById(Long id) {
        // 优先查询MySQL，当不存在时再查询ES | 大部分查询都是围绕当前热点数据（1个月内）查询，所以绝大多数情况会命中
        // 不直接走ES查询的原因是ES的写入可能存在延迟，有可能写入后立刻查询会查不到，且ES写入后续有可能改为异步写入（MQ）
        SecondCollection entity = baseDao.getById(id);
        if (entity != null) {
            return entity;
        }

        return esDao.findById(id);
    }

    /**
     * SecondCollection 事务事件
     */
    @Getter
    public static class SecondCollectionTxEvent extends ApplicationEvent {

        private final Long pk;

        public SecondCollectionTxEvent(long pk) {
            super(pk);
            this.pk = pk;
        }
    }

    /**
     * 同步ES事件
     */
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT, classes = SecondCollectionTxEvent.class)
    void onTxEvent(SecondCollectionTxEvent event) {
        Long id = event.getPk();
        SecondCollection entity = getById(id);
        if (entity != null) {
            esDao.save(entity);
        }
    }

    @Transactional
    public boolean save(SecondCollection entity) {
        boolean ret = baseDao.save(entity);
        if (ret) {
            publisher.publishEvent(new SecondCollectionTxEvent(entity.getId()));
        }
        return ret;
    }

    public boolean updateById(SecondCollection entity) {
        boolean ret = baseDao.updateById(entity);
        if (ret) {
            publisher.publishEvent(new SecondCollectionTxEvent(entity.getId()));
        }
        return ret;
    }

    @Repository
    static class SecondCollectionDaoAdapter extends ServiceImpl<SecondCollectionMapper, SecondCollection> {
        // 为 SecondCollectionDao 断舍离~
    }

    public SecondCollection getOnShelfUserCollectionId(Long userCollectionId) {
        // 默认条件 status IN (1, 4) 走MySQL查询
        LambdaQueryWrapper<SecondCollection> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(SecondCollection::getUserCollectionId, userCollectionId);
        wrapper.in(SecondCollection::getStatus, ConSignStatus.ON_SALES, ConSignStatus.LOCK);
        return baseDao.getOne(wrapper);
    }

    public Integer getOnShelfByUserIdAndCollectionIdCount(Long userId, Long collectionId) {
        // 默认条件 status IN (1, 4) 走MySQL查询
        LambdaQueryWrapper<SecondCollection> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(SecondCollection::getUserId, userId);
        wrapper.eq(SecondCollection::getCollectionId, collectionId);
        wrapper.in(SecondCollection::getStatus, ConSignStatus.ON_SALES, ConSignStatus.LOCK);
        return baseDao.count(wrapper);
    }

    public Boolean whetherAnItemIsOnConsignment(Long collectionId) {
        // 默认条件 status = 1 走MySQL查询
        LambdaQueryWrapper<SecondCollection> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(SecondCollection::getCollectionId, collectionId);
        wrapper.eq(SecondCollection::getStatus, ConSignStatus.ON_SALES);
        return baseDao.count(wrapper) > 0;
    }

    public boolean checkSelfSalesCollection(Long collectionId, Long userId) {
        return esDao.countSelfSalesCollection(collectionId, userId) > 0;
    }

    @Deprecated
    boolean checkSelfSalesCollection0(Long collectionId, Long userId) {
        LambdaQueryWrapper<SecondCollection> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(SecondCollection::getCollectionId, collectionId);
        wrapper.eq(SecondCollection::getUserId, userId);
        return baseDao.count(wrapper) > 0;
    }

    /**
     * 玛特宇宙后台 > 寄售市场 > 寄售列表查询 | MySQL侧只保留近30天和 status IN (1, 4) 的数据，其它数据走ES查询
     */
    @SneakyThrows
    public Page<SecondConSignCollectionDTO> getAdminConsignPage(Date startTime, Date endTime, String phone, ConSignStatus status, int pageNo, int pageSize, Integer marketType) {
        // status IN (1, 4) 或查询区间在30天内的走实时查询 | 主表保留了最近 30 天及 status IN (1, 4) 的数据
        if (status == ConSignStatus.ON_SALES || status == ConSignStatus.LOCK || isStartTimeIn30Days(startTime)) {
            long count = countAdminConsignPageFromMySQL(startTime, endTime, phone, status, marketType);
            List<SecondConSignCollectionDTO> list = null;
            if (count > 0) {
                list = getAdminConsignPageFromMySQL(startTime, endTime, phone, status, pageNo, pageSize, marketType);
            }
            return buildGetAdminConsignPageResult(count, list, pageNo, pageSize);
        }

        // 30天外查询完全走ES查询
        if (isEndTimeNotIn30Days(endTime)) {
            long count = esDao.countAdminConsignPage(startTime, endTime, phone, status, marketType);
            List<SecondConSignCollectionDTO> list = null;
            if (count > 0) {
                list = esDao.getAdminConsignPage(startTime, endTime, phone, status, pageNo, pageSize, marketType);
            }
            return buildGetAdminConsignPageResult(count, list, pageNo, pageSize);
        }

        // 如果时间范围不明确或者超过30天（开始时间），那么组合MySQL和ES数据查询
        // 将查询时间按30天的临界值拆分开 [startTime, 30天前], [30天前, endTime]
        Date esStartTime = startTime;
        Date esEndTime = DateUtil.offset(new Date(), DateField.DAY_OF_MONTH, -30);
        Date sqlStartTime = esEndTime;
        Date sqlEndTime = endTime;

        long sqlCount = countAdminConsignPageFromMySQL(sqlStartTime, sqlEndTime, phone, status, marketType);
        long esCount = esDao.countAdminConsignPage(esStartTime, esEndTime, phone, status, marketType);
        long count = sqlCount + esCount;

        // 根据查询页码和分页大小计算MySQL侧是否足页

        // MySQL 侧足页，直接返回
        if ((long) pageNo * pageSize <= sqlCount) {
            List<SecondConSignCollectionDTO> list = getAdminConsignPageFromMySQL(sqlStartTime, sqlEndTime, phone, status, pageNo, pageSize, marketType);
            return buildGetAdminConsignPageResult(count, list, pageNo, pageSize);
        }

        // MySQL侧已取空，完全走ES查询

        // 计算ES侧分页页码
        int esPageNo = (pageNo * pageSize - (int) sqlCount) / pageSize + 1;
        if (esCount > 0 && ((long) (pageNo - 1) * pageSize >= sqlCount)) {
            List<SecondConSignCollectionDTO> esList = esDao.getAdminConsignPage(esStartTime, esEndTime, phone, status, esPageNo, pageSize, marketType);
            return buildGetAdminConsignPageResult(count, esList, pageNo, pageSize);
        }

        // MySQL + ES 临界页，取双方数据补齐
        List<SecondConSignCollectionDTO> sqlList = getAdminConsignPageFromMySQL(sqlStartTime, sqlEndTime, phone, status, pageNo, pageSize, marketType);
        List<SecondConSignCollectionDTO> esList = esDao.getAdminConsignPage(esStartTime, esEndTime, phone, status, esPageNo, pageSize - sqlList.size(), marketType);

        List<SecondConSignCollectionDTO> list = new ArrayList<>();
        list.addAll(sqlList);
        list.addAll(esList);

        return buildGetAdminConsignPageResult(count, list, pageNo, pageSize);
    }

    private Page<SecondConSignCollectionDTO> buildGetAdminConsignPageResult(long count, List<SecondConSignCollectionDTO> list, int pageNo, int pageSize) {
        Page<SecondConSignCollectionDTO> page = new Page<>(pageNo, pageSize);
        page.setTotal(count);
        page.setRecords(list);
        return page;
    }

    /**
     * 30天对应的毫秒数
     */
    private static final long DURING_30_DAY_MILLIS = 30 * 24 * 3600 * 1000L;

    /**
     * 查询开始时间是否在30天内
     */
    private boolean isStartTimeIn30Days(Date startTime) {
        return startTime != null && DateTime.now().getTime() - startTime.getTime() < DURING_30_DAY_MILLIS;
    }

    /**
     * 查询结束时间是否不在30天内，是则完全走ES查询，否则将拆分查询
     */
    private boolean isEndTimeNotIn30Days(Date endTime) {
        return endTime != null && DateTime.now().getTime() - endTime.getTime() > DURING_30_DAY_MILLIS;
    }

    long countAdminConsignPageFromMySQL(Date startTime, Date endTime, String phone, ConSignStatus status, Integer marketType) {
        return baseMapper.getAdminConsignPageTol(startTime, endTime, status != null ? status.getCode() : null, phone, marketType);
    }

    List<SecondConSignCollectionDTO> getAdminConsignPageFromMySQL(Date startTime, Date endTime, String phone, ConSignStatus status, int pageNo, int pageSize, Integer marketType) {
        Page<?> page = new Page<>(pageNo, pageSize).setSearchCount(Boolean.FALSE);
        Page<SecondConSignCollectionDTO> pageResult = baseMapper.getAdminConsignPage(page, startTime, endTime, status != null ? status.getCode() : null, phone, marketType);
        return pageResult.getRecords();
    }

    public BigDecimal goodsTotalMoney(List<Long> ids) {
        List<SecondCollection> secondCollections = esDao.findByIds(ids);
        return secondCollections.stream().map(SecondCollection::getSalesPrice).reduce(BigDecimal.ZERO, BigDecimal::add);
    }

    @Deprecated
    BigDecimal goodsTotalMoney0(List<Long> ids) {
        List<SecondCollection> secondCollections = baseMapper.selectBatchIds(ids);
        return secondCollections.stream().map(SecondCollection::getSalesPrice).reduce(BigDecimal.ZERO, BigDecimal::add);
    }

    public Page<MaintainUserConSignCollectionPageDTO> maintainUserConsignPage(int pageNo, int pageSize,
                                                                              Long userId, BigDecimal lowPrice, BigDecimal highPrice, Long CollectionId) {
        // 默认条件 status = 1 走 MySQL 查询
        return baseMapper.maintainUserConsignPage(new Page<>(pageNo, pageSize), ConSignStatus.ON_SALES.getCode(),
                userId, lowPrice, highPrice, CollectionId);
    }


    public Integer presentCount(UserInfo userInfo, Long collectionId, Long maxId) {
        // 默认条件 status = 1 走 MySQL 查询
        return baseMapper.presentCount(collectionId, userInfo.getId(), maxId);
    }

    public SecondCollection selectByUserCollectionIdAndUserId(Long userCollectionId, Long userId) {
        return esDao.selectByUserCollectionIdAndUserId(userCollectionId, userId, ConSignStatus.CANCEL);
    }

    @Deprecated
    SecondCollection selectByUserCollectionIdAndUserId0(Long userCollectionId, Long userId) {
        QueryWrapper<SecondCollection> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("user_collection_id", userCollectionId);
        queryWrapper.eq("user_id", userId);
        queryWrapper.ne("status", ConSignStatus.CANCEL);
        return baseDao.getOne(queryWrapper);
    }

    public Page<MarketConsignPageDTO> marketConsignPage(Long collectionId, Integer sortType, Integer pageNo, Integer pageSize) {
        // 默认条件 status = 1 走MySQL查询
        return baseMapper.marketConsignPage(new Page(pageNo, pageSize), collectionId, sortType);
    }

    @Transactional
    public Boolean collectionSoldOut(Long id) {
        return resetStatus(id, ConSignStatus.ON_SALES, ConSignStatus.CANCEL);
    }

    public Page<MyConsignmentListVO> myConsignmentList(Page page, Long userId) {
        // 默认条件 status = 1 走MySQL查询
        return baseMapper.myConsignmentList(page, userId);
    }

    public Page<MySalestListVO> mySalestList(Page page, Long userId) {
        // 默认条件 status IN (1, 4) 走MySQL查询
        return baseMapper.mySalestList(page, userId);
    }

    public List<SecondCollection> offShelfAllList() {
        // 默认条件 status = 1 走MySQL查询
        LambdaQueryWrapper<SecondCollection> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(SecondCollection::getStatus, ConSignStatus.ON_SALES);
        wrapper.eq(SecondCollection::getMarketType, MarketTypeEnum.BRAND_INCUBATION);
        wrapper.lt(SecondCollection::getOffShelfTime, DateTime.now());
        return baseDao.list(wrapper);
    }

    public List<SecondCollection> getListByCollectionIdWithOnSales(Long collectionId) {
        // 默认条件 status = 1 走MySQL查询
        LambdaQueryWrapper<SecondCollection> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(SecondCollection::getCollectionId, collectionId);
        wrapper.eq(SecondCollection::getStatus, ConSignStatus.ON_SALES);
        return baseDao.list(wrapper);
    }

    public Page<MarketConsignPageV3_7VO> marketConsignPagev3_7(Long shareUserId, Long userId, Long collectionId, Integer sortType, Integer pageNo, Integer pageSize) {
        DigitalCollection digitalCollection = digitalCollectionDao.getEcologyCollectionById(collectionId);
        if (ObjectUtil.isNotNull(digitalCollection)) {
            ExitConsensusUser exitConsensusUser = exitConsensusUserDao.getByUserId(userId);
            if (ObjectUtil.isNotNull(exitConsensusUser) && exitConsensusUser.getStatus()) {
                // 默认条件 status IN (1, 4) 走MySQL查询
                return baseMapper.marketConsignPageForConsensus(new Page(pageNo, pageSize), collectionId, sortType, shareUserId);
            } else {
                // 默认条件 status IN (1, 4) 走MySQL查询
                return baseMapper.marketConsignPageNotInConsensus(new Page(pageNo, pageSize), collectionId, sortType, shareUserId);

            }
        }
        // 默认条件 status IN (1, 4) 走MySQL查询
        return baseMapper.marketConsignPagev3_7(new Page(pageNo, pageSize), collectionId, sortType, shareUserId);
    }

    public Page<MarketConsignPageV3_7VO> marketConsignPagev3_7ForMarket(Long shareUserId, Long collectionId, Integer sortType, Integer pageNo, Integer pageSize) {
        // 默认条件 status IN (1, 4) 走MySQL查询
        return baseMapper.marketConsignPagev3_7ForMarket(new Page(pageNo, pageSize), collectionId, sortType, shareUserId);
    }

    public ConsignmentSalesIdGetParticularsVO consignmentSalesIdGetParticulars(Long id) {
        SecondCollection entity = this.getById(id);
        if (entity == null) {
            return null;
        }

        // 组装字段
        ConsignmentSalesIdGetParticularsVO vo = new ConsignmentSalesIdGetParticularsVO();
        vo.setNumbering(String.valueOf(entity.getNumbering()));
        // join sc_digital_collection(collection_id > first_title)
        DigitalCollection digitalCollection = digitalCollectionDao.getById(entity.getCollectionId());
        if (digitalCollection != null) {
            vo.setFirstTitle(digitalCollection.getFirstTitle());
        }
        // join sc_user_collection(user_collection_id > chain_url)
        UserCollection userCollection = userCollectionDao.getById(entity.getUserCollectionId());
        if (userCollection != null) {
            vo.setHoldContractAddress(userCollection.getChainUrl());
        }
        return vo;
    }

    @Deprecated
    ConsignmentSalesIdGetParticularsVO consignmentSalesIdGetParticulars0(Long id) {
        return baseMapper.consignmentSalesIdGetParticulars(id);
    }

    public Long getMaxId() {
        // 走MySQL查询
        return baseMapper.getMaxId();
    }

    public Long getYesterdayMaxId() {
        // 走MySQL查询
        QueryWrapper<SecondCollection> queryWrapper = new QueryWrapper<>();
        queryWrapper.select("id");
        queryWrapper.lt("create_time", DateUtil.beginOfDay(DateTime.now()));
        queryWrapper.orderByDesc("id");
        queryWrapper.last("limit 1");
        return baseDao.getObj(queryWrapper, x -> (Long) x);
    }

    public List<SecondCollection> selectByCollectionIdAndMarketType(Long collectionId, MarketTypeEnum marketTypeEnum) {
        // 默认条件 status = 1 走MySQL查询
        LambdaQueryWrapper<SecondCollection> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(SecondCollection::getCollectionId, collectionId);
        wrapper.eq(SecondCollection::getMarketType, marketTypeEnum);
        wrapper.eq(SecondCollection::getStatus, ConSignStatus.ON_SALES);
        return baseDao.list(wrapper);
    }

    public List<Long> getByOnSalesIdWithOutSelf(MarketTypeEnum marketType, Long collectionId, Long userId, Long maxId, Integer count) {
        // 默认条件 status = 1 走MySQL查询
        LambdaQueryWrapper<SecondCollection> wrapper = new LambdaQueryWrapper<>();
        wrapper.select(SecondCollection::getId);
        wrapper.eq(SecondCollection::getCollectionId, collectionId);
        wrapper.eq(SecondCollection::getMarketType, marketType);
        wrapper.ne(SecondCollection::getUserId, userId);
        wrapper.gt(SecondCollection::getId, maxId);
        wrapper.eq(SecondCollection::getStatus, ConSignStatus.ON_SALES);
        wrapper.last(" limit " + count);
        return baseDao.listObjs(wrapper, (x) -> (Long) x);
    }

    /**
     * 重置状态
     */
    boolean resetStatus(@NonNull Long id, @NonNull ConSignStatus originStatus, @NonNull ConSignStatus targetStatus) {
        if (originStatus == targetStatus) {
            return false;
        }
        LambdaUpdateWrapper<SecondCollection> wrapper = new LambdaUpdateWrapper<>();
        wrapper.eq(SecondCollection::getId, id);
        wrapper.eq(SecondCollection::getStatus, originStatus);
        wrapper.set(SecondCollection::getStatus, targetStatus);
        boolean ret = baseDao.update(wrapper);
        // 同步更新ES
        if (ret) {
//            SecondCollection entity = new SecondCollection();
//            entity.setId(id);
//            entity.setStatus(targetStatus);
//            updateToEs(entity);
            publisher.publishEvent(new SecondCollectionTxEvent(id));
        }
        return ret;
    }

    @Transactional
    public boolean cancelOrder(Long id) {
        return resetStatus(id, ConSignStatus.LOCK, ConSignStatus.ON_SALES);
    }

    @Transactional
    public boolean lockOrder(Long id) {
        return resetStatus(id, ConSignStatus.ON_SALES, ConSignStatus.LOCK);
    }

    public String getMinPriceByCollectionId(Long collectionId) {
        String key = RedisKeyEnum.FREE_MARKET_LOWEST_PRICE.getKey() + collectionId;
        Boolean b = redisTemplate.hasKey(key);
        if (b) {
            return (String) redisTemplate.opsForValue().get(key);
        }
        // 默认条件 status = 1 走MySQL查询
        BigDecimal minPriceByCollectionId = baseMapper.getMinPriceByCollectionId(collectionId);
        String plainString = minPriceByCollectionId.setScale(2, BigDecimal.ROUND_DOWN).toPlainString();
        redisTemplate.opsForValue().set(key, plainString, 2, TimeUnit.MINUTES);
        return plainString;
    }

    public String getSalesCountByCollectionId(Long collectionId) {
        String key = RedisKeyEnum.SALES_COUNT.getKey() + collectionId;
        Boolean b = redisTemplate.hasKey(key);
        if (b) {
            return (String) redisTemplate.opsForValue().get(key);
        }
        // 默认条件 status = 1 走MySQL查询
        String salesCountByCollectionId = baseMapper.getSalesCountByCollectionId(collectionId);
        redisTemplate.opsForValue().set(key, salesCountByCollectionId, 2, TimeUnit.MINUTES);
        return salesCountByCollectionId;
    }

    public String getSalesCountByCollectionIdForConsensus(Long collectionId) {
        String key = RedisKeyEnum.SALES_CONSENSUS_COUNT.getKey() + collectionId;
        Boolean b = redisTemplate.hasKey(key);
        if (b) {
            return (String) redisTemplate.opsForValue().get(key);
        }
        // 默认条件 status = 1 走MySQL查询
        String salesCountByCollectionId = baseMapper.getSalesCountByCollectionIdForConsensus(collectionId);
        redisTemplate.opsForValue().set(key, salesCountByCollectionId, 2, TimeUnit.MINUTES);
        return salesCountByCollectionId;
    }

    public String getSalesCountByCollectionIdNotInConsensus(Long collectionId) {
        String key = RedisKeyEnum.SALES_COUNT.getKey() + collectionId;
        Boolean b = redisTemplate.hasKey(key);
        if (b) {
            return (String) redisTemplate.opsForValue().get(key);
        }
        // 默认条件 status = 1 走MySQL查询
        String salesCountByCollectionId = baseMapper.getSalesCountByCollectionIdNotInConsensus(collectionId);
        redisTemplate.opsForValue().set(key, salesCountByCollectionId, 2, TimeUnit.MINUTES);
        return salesCountByCollectionId;
    }

    public SecondCollection findById(Long id) {
        SecondCollection entity = baseMapper.selectById(id);
        if (entity != null) {
            return entity;
        }
        return esDao.findById(id);
    }

    /**
     * 后台市值账号批量购买自由市场藏品专用 禁止他用
     *
     * @param userId
     * @return
     */
    public List<Long> getCollectionIdsByMaintainUser(Long userId, Long collectionId, Integer num, BigDecimal lowPrice, BigDecimal highPrice) {
        // 默认条件 status = 1 走MySQL查询
        return baseMapper.findMaintainUserLowPriceConsign(userId, collectionId, num, lowPrice, highPrice);
    }

    public List<MySalestListVO> maintainSalestList(Long userId, Long collectionId, BigDecimal lowPrice, BigDecimal highPrice) {
        // 默认条件 status IN (1, 4) 走MySQL查询
        return baseMapper.maintainSalestList(userId, collectionId, lowPrice, highPrice);
    }

    public BigDecimal getCollectionLowestPrice(Long collectionId) {
        // 默认条件 status = 1 走MySQL查询
        return baseMapper.getCollectionLowestPrice(collectionId);
    }

    public BigDecimal getCollectionLowestPriceForConsensus(Long collectionId) {
        // 默认条件 status = 1 走MySQL查询
        return baseMapper.getCollectionLowestPriceForConsensus(collectionId);
    }

    public BigDecimal getCollectionLowestPriceNotInConsensus(Long collectionId) {
        // 默认条件 status = 1 走MySQL查询
        return baseMapper.getCollectionLowestPriceNotInConsensus(collectionId);
    }

    public BigDecimal getConsensusMinPriceByCollectionId(Long collectionId) {
        // 默认条件 status IN (1, 4) 走MySQL查询
        return baseMapper.getConsensusMinPriceByCollectionId(collectionId);
    }

    public Integer getConsensusCountByCollectionIdAndPrice(Long collectionId, BigDecimal price) {
        // 默认条件 status IN (1, 4) 走MySQL查询
        return baseMapper.getConsensusCountByCollectionIdAndPrice(collectionId, price);
    }

    /**
     * 滚动同步MySQL数据到ES
     */
    public void syncToEs(Long collectionId, String userTel) {

        final String key = String.format("data:sync:sc_second_collection:%d:user:%s", collectionId, userTel);

        if (!redisTemplate.opsForValue().setIfAbsent(key, "1", Duration.ofMinutes(5))) {
            log.info("表 sc_second_collection 数据同步中 collectionId: {}, userTel: {}", collectionId, userTel);
            return;
        }

        Long userId = null;
        if (StrUtil.isNotBlank(userTel)) {
            UserInfo user = userInfoDao.getByUserTel(userTel);
            if (user != null) {
                userId = user.getId();
            }
        }

        if (collectionId == null && userId == null) {
            log.warn("请求参数错误（藏品或手机号不能同时为空）");
            redisTemplate.delete(key);
            return;
        }

        log.info("表 sc_second_collection 数据开始同步 collectionId: {}, userTel: {}", collectionId, userTel);

        // 遍历 sc_second_collection 数据
        LambdaQueryWrapper<SecondCollection> wrapper = Wrappers.lambdaQuery();
        wrapper.eq(collectionId != null, SecondCollection::getCollectionId, collectionId);
        wrapper.eq(userId != null, SecondCollection::getUserId, userId);

        int pageNo = 1;
        int limit = 1000;
        for (; ; pageNo++) {
            // 分页查询
            Page<SecondCollection> page = baseDao.page(new Page<>(pageNo, limit, false), wrapper);
            if (CollectionUtils.isEmpty(page.getRecords())) {
                break;
            }

            // 执行同步
            try {
                esDao.bulkIndex(page.getRecords());
            } catch (IOException e) {
                log.error("同步ES数据出错", e);
            }

            if (page.getRecords().size() < limit) {
                break;
            }
        }

        redisTemplate.delete(key);
    }

}